package com.neusoft.bd.myspark.spark01

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object M20AggregateByKey {

  //  相同key的平均值
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc: SparkContext = new SparkContext(conf)
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)), 2)
//    直接将第一个值作为初始值
    rdd.combineByKey(
//      第一个值不做计算，本题只转换结构
      v => (v,1),
      (t:(Int,Int),v)=>{ (t._1+v , t._2+1) },
      (t1:(Int,Int),t2:(Int,Int)) => { (t1._1 +t2._1 , t2._2 + t2._2 )}
    ).mapValues{
      case (num , cnt)=> num /cnt
    }
      .collect().foreach(println)


  }

  //  相同key的平均值
  def main4(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc: SparkContext = new SparkContext(conf)
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)), 2)
    //    分区内和分区间计算相同
    val rddt: RDD[(String, (Int, Int))] = rdd.aggregateByKey(
      //      初始值：一个tuple对象， 定义为： _1求和，_2统计个数
      (0, 0))(
      //      分区内
      (t, v) => {
        (t._1 + v, t._2 + 1)
      }
      ,
      //
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2)
      }
    )
    //    key 保存不变，但是V进行计算
    val unit: RDD[(String, Int)] = rddt.mapValues {
      case (num, cnt) => num / cnt
    }
    unit.collect().foreach(println)

  }

  def main3(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc: SparkContext = new SparkContext(conf)
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)), 2)
    //    分区内和分区间计算相同
    rdd.foldByKey(0)(_ + _).collect().foreach(println)
  }

  def main2(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc: SparkContext = new SparkContext(conf)
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)), 2)
    rdd.aggregateByKey(
      //      分区间比较时需要的初始值
      //      初始值绝对结果类型决定最终返回结果
      Long.MinValue)(
      //      分区内取最大值
      //      1 分区： a(2) , b(3)
      //      2分区： a（6） , b(5)
      (x, y) => math.max(x, y),
      //      分区间求和
      (x, y) => x + y
    ).collect().foreach(println)

  }


  def main1(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc: SparkContext = new SparkContext(conf)
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("b", 4), ("b", 5), ("b", 6)), 2)
    //    函数柯里化
    //


    rdd.aggregateByKey(
      //      分区间比较时需要的初始值
      Long.MinValue)(
      //      分区内取最大值
      (x, y) => math.max(x, y),
      //      分区间求和
      (x, y) => x + y
    ).collect().foreach(println)

    print(123)
  }

}
